﻿import 'dart:async';
import 'dart:io';

import 'package:get/get.dart';
import 'package:influxdb_client/api.dart';
import 'package:mqtt_client/mqtt_client.dart';
import 'package:mqtt_client/mqtt_server_client.dart';
import 'package:simudata_display/app/data/protobuf/simu_message.pb.dart';
import 'package:simudata_display/app/services/influx_db_client_service.dart';

class MqttClientService extends GetxService {
  MqttServerClient? _client;

  var messageStream = StreamController<SimuMessage>.broadcast();

  final int limitMessageLength = 3000;

  Map<String, Map<String, List<SimuData>>> simuDataMap = {};

  bool isConnected() {
    if (null == _client) {
      return false;
    }
    return _client!.connectionStatus!.state == MqttConnectionState.connected;
  }

  Future<bool> connectToServer(String serverIP, String clientID) async {
    if (isConnected()) {
      return false;
    }
    _client ??= MqttServerClient(serverIP, clientID);
    try {
      await _client!.connect();
    } on NoConnectionException catch (e) {
      _client!.disconnect();
    } on SocketException catch (e) {
      _client!.disconnect();
    }
    return isConnected();
  }

  bool subscribe(String topic, int qos) {
    if (isConnected()) {
      _client!.subscribe(topic, MqttQos.values[qos]);
      _client!.updates!.listen(_onReceiveMessage);
      return true;
    }

    return false;
  }

  void _onReceiveMessage(List<MqttReceivedMessage<MqttMessage>> messages) {
    InfluxDBClientService dbClientService = Get.find();
    for (var msg in messages) {
      final recvMsg = msg.payload as MqttPublishMessage;
      SimuMessage simuMessage = SimuMessage.fromBuffer(recvMsg.payload.message);
      for (var data in simuMessage.datas) {
        String sersorName = data.id.toString();
        // 获取influxdb的数据写入接口，如果接口不为null说明writeService可用
        var writeService = dbClientService.getDBWriteService();
        if (writeService != null) {
          // 生成point数据
          var influxdbPoint = Point(simuMessage.name)
              .addTag("sensor", sersorName)
              .addField("value", data.value)
              .time(data.timestamp);
          writeService.batchWrite(influxdbPoint);
        }

        if (simuDataMap.containsKey(simuMessage.name)) {
          simuDataMap[simuMessage.name]?.update(sersorName, (value) {
            if (value.length > limitMessageLength) {
              value.removeAt(0);
            }
            value.add(data);
            return value;
          }, ifAbsent: () => <SimuData>[data]);
        } else {
          Map<String, List<SimuData>> deviceMap = {};
          deviceMap[sersorName] = <SimuData>[data];
          simuDataMap[simuMessage.name] = deviceMap;
        }
      }
      messageStream.add(simuMessage);
    }
  }

  @override
  void onInit() {
    super.onInit();
  }

  @override
  void onReady() {
    super.onReady();
  }

  @override
  void onClose() {
    if (isConnected()) {
      _client!.disconnect();
    }
    super.onClose();
  }
}
